package com.learn.window.aggregation;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;

/**
 * @create: 2023-04-23 21:57
 * @author: Mr.Du
 * --------------
 * @notes: 测试数据: 某个用户在某个时刻浏览了某个商品，以及商品的价值
 * {"userID": "user_4", "eventTime": "2019-11-09 10:41:32", "eventType": "browse", "productID": "product_1", "productPrice": 10}
 * API: 如上AggregateFunction与ProcessWindowFunction
 * 示例: 获取一段时间内(Window Size)每个用户(KeyBy)浏览的平均价值(AggregateFunction),并获得Key和Window信息。
 **/
public class AggregateAndProcessFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        env.setParallelism(1);
        DataStreamSource<String> streamSource = env.socketTextStream("node1", 9999);
        
        streamSource.map(new MapFunction<String, UserActionLog1>() {
            @Override
            public UserActionLog1 map(String s) throws Exception {
                return JSON.parseObject(s, UserActionLog1.class);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<UserActionLog1>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner(new SerializableTimestampAssigner<UserActionLog1>() {
                    @Override
                    public long extractTimestamp(UserActionLog1 userActionLog1, long recordTimestamp) {
                        try {
                            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                            return format.parse(userActionLog1.getEventTime()).getTime();
                        } catch (ParseException e) {
                            e.printStackTrace();
                            return 0L;
                        }
                    }
                }))
                //按照用户ID分组
                .keyBy(UserActionLog1::getUserID)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .aggregate(new AggregateFunction<UserActionLog1, Tuple2<Long, Long>, Double>() {

                    /**
                     * 1、初始值
                     * //定义累加器初始值
                     *
                     * @return
                     */
                    @Override
                    public Tuple2<Long, Long> createAccumulator() {
                        return new Tuple2<>(0L, 0L);
                    }

                    /**
                     * 2、累加
                     * 定义累加器如何基于输入数据进行累加
                     *
                     * @param value
                     * @param accumulator
                     * @return
                     */
                    @Override
                    public Tuple2<Long, Long> add(UserActionLog1 value, Tuple2<Long, Long> accumulator) {
                        accumulator.f0 += 1;
                        accumulator.f1 += value.getProductPrice();
                        return accumulator;
                    }

                    /**
                     * 3、合并
                     * 定义累加器如何和State中的累加器进行合并
                     *
                     * @param acc1
                     * @param acc2
                     * @return
                     */
                    @Override
                    public Tuple2<Long, Long> merge(Tuple2<Long, Long> acc1, Tuple2<Long, Long> acc2) {
                        acc1.f0 += acc2.f0;
                        acc1.f1 += acc2.f1;
                        return acc1;
                    }

                    /**
                     * 4、输出
                     * 定义如何输出数据
                     *
                     * @param accumulator
                     * @return
                     */
                    @Override
                    public Double getResult(Tuple2<Long, Long> accumulator) {
                        return accumulator.f1 / (accumulator.f0 * 1.0);
                    }
                }, new ProcessWindowFunction<Double, String, String, TimeWindow>() {
                    @Override
                    public void process(String key, ProcessWindowFunction<Double, String, String, TimeWindow>.Context context, Iterable<Double> iterable, Collector<String> out) throws Exception {
                        Double avg = iterable.iterator().next();
                        String windowStart = new DateTime(context.window().getStart(), DateTimeZone.forID("+08:00")).toString("yyyy-MM-dd HH:mm:ss");
                        String windowEnd = new DateTime(context.window().getEnd(), DateTimeZone.forID("+08:00")).toString("yyyy-MM-dd HH:mm:ss");
                        String record="Key: "+key+" 窗口开始时间: "+windowStart+" 窗口结束时间: "+windowEnd+" 浏览的商品的平均价值: "+String.format("%.2f",avg);
                        out.collect(record);
                    }
                }).print();
        env.execute();
    }
}


@Data
@AllArgsConstructor
@NoArgsConstructor
class UserActionLog1{
    private String userID;
    private String eventTime;
    private String eventType;
    private String productID;
    private Long productPrice;
}
